package org.example.statistics

import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

import java.text.SimpleDateFormat
import java.util.Date

import org.example.scala.model.{MongoConfig, Movie, MovieRating, Recommendation, GenresRecommendation}
import org.example.java.model.Constant._

object StatisticsRecommender {

  //  val MONGODB_RATING_COLLECTION = "Rating"
  //  val MONGODB_MOVIE_COLLECTION = "Movie"

  //  统计的表的名称
  //  val RATE_MORE_MOVIES = "RateMoreMovies"
  //  val RATE_MORE_RECENTLY_MOVIES = "RateMoreRecentlyMovies"
  //  val AVERAGE_MOVIES = "AverageMovies"
  //  val GENRES_TOP_MOVIES = "GenresTopMovies"

  //  入口方法
  def main(args: Array[String]): Unit = {
    val config = Map(
      "spark.cores" -> "local[*]",
      "mongo.uri" -> "mongodb://localhost:27018/recommender",
      "mongo.db" -> "recommender"
    )

    //    创建SparkConf配置
    val sparkConf = new SparkConf().setAppName("StatisticsRecommender").setMaster(config("spark.cores"))

    //    创建SparkSession
    val spark = SparkSession.builder().config(sparkConf).getOrCreate()

    val mongoConfig = MongoConfig(config("mongo.uri"), config("mongo.db"))

    //    加入隐式转换
    import spark.implicits._

    //    数据加载进来
    val ratingDF = spark
      .read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGO_RATING_COLLECTION)
      .format(MONGO_DRIVER_CLASS)
      .load()
      .as[MovieRating]
      .toDF()
    val movieDF = spark
      .read
      .option("uri", mongoConfig.uri)
      .option("collection", MONGO_MOVIE_COLLECTION)
      .format(MONGO_DRIVER_CLASS)
      .load()
      .as[Movie]
      .toDF()

    //    创建一张叫ratings的表
    ratingDF.createOrReplaceTempView("ratings")

    //    1.统计所有历史数据中每个电影的评分数
    //    数据结构 mid,count
    val rateMoreMoviesDF = spark.sql("select mid, count(mid) as count from ratings group by mid")

    rateMoreMoviesDF
      .write
      .option("uri", mongoConfig.uri)
      .option("collection", MONGO_RATE_MORE_MOVIES)
      .mode("overwrite")
      .format(MONGO_DRIVER_CLASS)
      .save()

    //    2.统计以月为单位每个电影的评分数
    //    数据结构 mid,count,time(年月)
    val simpleDateFormat = new SimpleDateFormat("yyyyMM")
    //    注册一个UDF函数,用于将timestamp转换成年月格式
    spark.udf.register("changeDate", (x: Int) => simpleDateFormat.format(new Date(x * 1000L)).toInt)

    //    将原来数据集中的时间转换成年月格式
    val ratingOfYearMonth = spark.sql("select mid, score, changeDate(timestamp) as yearmonth from ratings")

    //    将新的数据集注册成一张新的表
    ratingOfYearMonth.createOrReplaceTempView("ratingOfMonth")

    val rateMoreRecentlyMovies = spark.sql("select mid, count(mid) as count, yearmonth from ratingOfMonth group by yearmonth, mid")

    rateMoreRecentlyMovies
      .write
      .option("uri", mongoConfig.uri)
      .option("collection", MONGO_RATE_MORE_RECENTLY_MOVIES)
      .mode("overwrite")
      .format(MONGO_DRIVER_CLASS)
      .save()

    //    3.统计每个电影的平均评分
    val averageMoviesDF = spark.sql("select mid, avg(score) as avg from ratings group by mid")

    averageMoviesDF
      .write
      .option("uri", mongoConfig.uri)
      .option("collection", MONGO_AVERAGE_MOVIES)
      .mode("overwrite")
      .format(MONGO_DRIVER_CLASS)
      .save()

    //    4.统计每种电影类别中评分最高的10个电影
    //    需要用left join,因为只需有评分的电影
    val movieWithScore = movieDF.join(averageMoviesDF, Seq("mid", "mid"))

    //所有的电影类别
    val genres = List("Action", "Adventure", "Animation", "Comedy", "Crime", "Documentary", "Drama", "Family", "Fantasy", "Foreign", "History", "Horror", "Music", "Mystery", "Romance", "Science", "Tv", "Thriller", "War", "Western"
    )

    //    将电影类别转换成RDD
    val genresRDD = spark.sparkContext.makeRDD(genres)

    //    计算电影类别Top10
    val genresTopMovies = genresRDD.cartesian(movieWithScore.rdd) //将电影类别和电影数据进行笛卡尔积操作
      .filter {
        //            过滤掉电影的类别不匹配
        case (genres, row) => row.getAs[String]("genres").toLowerCase.contains(genres.toLowerCase)
      }
      .map {
        //            将整个数据集的数据量减少，生成RDD[String, Iter[mid, avg]]
        case (genres, row) => {
          (genres, (row.getAs[Int]("mid"), row.getAs[Double]("avg")))
        }
      }.groupByKey() //将数据集中的相同的genres聚集
      .map {
        //            通过评分的大小进行数据的排序，将数据映射为对象
        case (genres, items) => GenresRecommendation(genres, items.toList.sortWith(_._2 > _._2).take(10).map(item => Recommendation(item._1, item._2)))
      }.toDF()

    //    输出数据到mongodb
    genresTopMovies
      .write
      .option("uri", mongoConfig.uri)
      .option("collection", MONGO_GENRES_TOP_MOVIES)
      .mode("overwrite")
      .format(MONGO_DRIVER_CLASS)
      .save()

    //    关闭Spark
    spark.stop()
  }
}
